package com.jacquesh.sink;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.util.Properties;

/**
 * @author Jacquesh
 * @version 1.0
 * @date 2021/6/6 14:41
 */
public class KafkaSinkDemo {
    public static void main(String[] args) {
        StreamExecutionEnvironment environment =
                StreamExecutionEnvironment.getExecutionEnvironment();
        /**设置默认并行度为4*/
        environment.setParallelism(4);
        /**获取随机数流 ： 1~10*/
        DataStreamSource<Long> stream = environment.generateSequence(1,10);
        stream.flatMap(new FlatMapFunction<Long, String>() {
            @Override
            public void flatMap(Long aLong, Collector<String> collector) throws Exception {
                try{
                    collector.collect(aLong.toString());
                }catch (Exception ex){

                }
            }
        }).addSink(new FlinkKafkaProducer<String>("192.168.150.110:9092",
                "demo",
                new SimpleStringSchema()));
    }
}
